package edu.utexas.mapreduce;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;

public class InputSplitterLine implements InputSplitter {

	final int blockSize;
	InputSplitterLine(int blockSize) {
		this.blockSize = blockSize;
	}
	
	public List<InputBlock> splitInput(File inputFile) {
		
		try {
			BufferedReader fileReader = new BufferedReader(new InputStreamReader(new FileInputStream(inputFile), "UTF-8"));
			String fileName = inputFile.getAbsolutePath();
			long fileLength = inputFile.length();
			long blockLength = this.blockSize;
			List<InputBlock> blockList = new ArrayList<InputBlock>();
			long splitCount = (fileLength + blockLength - 1) / blockLength;
			
			long currentIndex = 0;
			long endIndex = 0;
			while(currentIndex < fileLength) {
				
				long startIndex = endIndex;
				currentIndex += fileReader.skip(blockLength);
				endIndex = currentIndex;
				
				// find the end of line
				while(true) {
					int ch = fileReader.read();
					
					// EOF.  do not advance the position
					if (ch == -1) {
						break;
					}
					
					currentIndex++; 
					endIndex++;
					
					// newline
					if (ch == '\n') {
						break;
					}
					
					// carriage return
					if (ch == '\r') {
						// check if the next byte is a newline.  
						ch = fileReader.read();
						currentIndex++;
						
						// if it is then it is part of the end-of-line terminator.
						// increment the end position
						if (ch == '\n') {
							endIndex++;
						}
						break;
					}	
				}
				
				blockList.add(new InputBlock(fileName, startIndex, endIndex));
			}
			
			// if there's a single split, we are done. 
			if (splitCount == 1) {
				return blockList;
			}

			// check if the last block is 'small' (half the block size)
			// if so, merge the last two blocks
			int lastIndex = blockList.size() - 1;
			InputBlock lastBlock = blockList.get(lastIndex);
			boolean mergeLast = (lastBlock.endOffset - lastBlock.startOffset) < (blockLength / 2);
			if (mergeLast) {
				InputBlock endBlock = blockList.remove(lastIndex);
				InputBlock startBlock = blockList.remove(lastIndex - 1);
				blockList.add(new InputBlock(fileName, startBlock.startOffset, endBlock.endOffset));
			}
			
			return blockList;
			
		} catch (FileNotFoundException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
		
		return new ArrayList<InputBlock>();
	}
}
